package com.zyx.flinkdemo.stream.sink;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zyx.flinkdemo.pojo.Nested;
import com.zyx.flinkdemo.pojo.Props;
import com.zyx.flinkdemo.stream.utils.ClickHouseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author zyx
 * @since 2021/7/16 21:45
 * desc: 使用ClickHouse的JDBC写入带有Nested结构的ClickHouse案例
 */
@Slf4j
public class ClickHouseNestedSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Props props = new Props();
        props.setPid("1002");
        props.setPnames("p1");
        JSONArray jsonArray = new JSONArray();
        jsonArray.add(props);
        JSONObject jsonObj1 = new JSONObject();
        jsonObj1.put("uid", "1001");
        jsonObj1.put("name", "zhangsan");
        jsonObj1.put("props", jsonArray);

        Nested nested1 = JSONObject.parseObject(jsonObj1.toJSONString(), Nested.class);

        JSONObject jsonObj2 = new JSONObject();
        jsonObj2.put("uid", "1001");
        //jsonObj2.put("name", "zhangsan");

        Nested nested2 = JSONObject.parseObject(jsonObj2.toJSONString(), Nested.class);


        env
                .fromElements(nested1, nested2)
                .addSink(ClickHouseUtil.getNestedJdbcSink("insert into test_nested values(?,?,?,?,?)"));

        env.execute();
    }
}
